package cn.iocoder.yudao.module.vos.job;

import cn.hutool.core.io.FileUtil;
import cn.hutool.json.JSONUtil;
import cn.iocoder.yudao.framework.quartz.core.handler.JobHandler;
import cn.iocoder.yudao.framework.tenant.core.aop.TenantIgnore;
import cn.iocoder.yudao.module.vos.dal.dataobject.ecdr.ECdrDOWithoutTag;
import cn.iocoder.yudao.module.vos.job.component.AiNlpClient;
import cn.iocoder.yudao.module.vos.service.ecdr.admin.ECdrService;
import cn.iocoder.yudao.module.vos.utils.CdrUtil;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.model.UpdateOptions;
import lombok.extern.slf4j.Slf4j;
import org.bson.Document;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.stereotype.Component;

// === SFTP ===
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.Session;
import com.jcraft.jsch.SftpATTRS;
import com.jcraft.jsch.SftpException;

import javax.annotation.Resource;
import java.io.File;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.time.Instant;
import java.time.LocalDate;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.ConcurrentHashMap;

@Component
@Slf4j
public class RecordingProcessingJob implements JobHandler {

    // 每完成多少条打印一次进度
    private static final int PROGRESS_LOG_EVERY = 50;

    // === 单机互斥运行标记（同一 JVM 内只允许一个实例在跑） ===
    private static final java.util.concurrent.atomic.AtomicBoolean RUNNING = new java.util.concurrent.atomic.AtomicBoolean(false);
    private static final AtomicLong LAST_START_AT = new AtomicLong(0L);

    // === SFTP 固定配置（22 端口） ===
    private static final String SFTP_HOST = "8.222.176.40";
    private static final int SFTP_PORT = 22;
    private static final String SFTP_USER = "root";
    private static final String SFTP_PASS = "qwerFDSA!23";

    // 远端目录（按天/小时分层）
    private static final String REMOTE_ROOT = "/kunshi-data/mp3";

    @Value("${recording.local-dir}")
    private String LOCAL_DIR;

    // 可配置并发度（默认 4）
    @Value("${recording.parallelism:4}")
    private int PARALLELISM;

    private static final ZoneId ZONE = ZoneId.systemDefault();
    private static final DateTimeFormatter DAY_FMT = DateTimeFormatter.ofPattern("yyyyMMdd");
    private static final DateTimeFormatter TS_FMT_PH = DateTimeFormatter.ofPattern("yyyyMMddHHmmss")
            .withZone(CdrUtil.ZONE);

    // Mongo 集合名
    private static final String MONGO_COLLECTION_NAME = "cdr_ai";

    @Resource
    private ECdrService cdrService;

    @Resource
    private AiNlpClient aiNlpClient;

    @Resource
    private MongoTemplate mongoTemplate;

    /**
     * ts(毫秒) → 菲律宾时间 yyyyMMddHHmmss
     */
    public static String formatToPhYmdHms(long ts) {
        return TS_FMT_PH.format(Instant.ofEpochMilli(ts));
    }

    /**
     * ts(毫秒) → 菲律宾时间 yyyyMMdd
     */
    private static String formatToPhYmd(long ts) {
        return DAY_FMT.format(Instant.ofEpochMilli(ts).atZone(CdrUtil.ZONE).toLocalDate());
    }

    @TenantIgnore
    @Override
    public String execute(String param) {
        // === 互斥检查 ===
        if (!RUNNING.compareAndSet(false, true)) {
            long last = LAST_START_AT.get();
            String lastStr = (last > 0) ? Instant.ofEpochMilli(last).atZone(ZONE).toString() : "N/A";
            log.warn("RecordingProcessingJob is already running. Skip this trigger. lastStartAt={}", lastStr);
            return "SKIPPED: already running";
        }
        LAST_START_AT.set(System.currentTimeMillis());
        long t0 = System.currentTimeMillis();

        try {
            RunParams rp = parseParams(param);
            log.info("[RecordingProcessingJob] params => skipDays={} holdSeconds={} limit={} appointedDate={}",
                    rp.skipDays, rp.holdSeconds, rp.limit, rp.appointedDate);

            final AtomicInteger ok = new AtomicInteger(0);
            final AtomicInteger fail = new AtomicInteger(0);
            final AtomicInteger done = new AtomicInteger(0); // 已完成（包含跳过、预处理失败、任务执行完成）

            // 本次运行内去重集：避免同一 (phone, dayKey) 在并发下重复处理
            final ConcurrentHashMap.KeySetView<String, Boolean> inRunSeen = ConcurrentHashMap.newKeySet();

            // 与服务侧一致：若传 appointedDate 则取 appointedDate-1，否则昨天
            String remoteDay = resolveTargetDayForBaseDir(rp.appointedDate);
            String remoteBase = REMOTE_ROOT + "/" + remoteDay;

            // 本地目录：仅 mp3（转写文本与JSON都不落盘）
            Path mp3Dir = Paths.get(LOCAL_DIR, "mp3");
            ensureLocalDir(mp3Dir.toString());

            // 拉取候选 CDR（limit 为 null 或 <=0 时不加 LIMIT）
            Integer queryLimit = (rp.limit > 0) ? Integer.valueOf(rp.limit) : null;
            List<ECdrDOWithoutTag> list =
                    cdrService.listLastDayCdrByHoldTime(rp.holdSeconds, queryLimit, rp.appointedDate);

            final int total = (list == null) ? 0 : list.size();
            log.info("Candidates fetched. total={}", total);

            // 线程池：有界队列，避免把外部服务打爆
            ThreadFactory tf = r -> {
                Thread t = new Thread(r);
                t.setName("rec-proc-" + t.getId());
                t.setDaemon(true);
                return t;
            };
            ExecutorService pool = new ThreadPoolExecutor(
                    Math.max(1, PARALLELISM), Math.max(1, PARALLELISM),
                    60L, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(Math.max(32, Math.max(1, PARALLELISM) * 8)),
                    tf,
                    new ThreadPoolExecutor.CallerRunsPolicy() // 队列满时回退到调用线程执行，起到自限流
            );

            Session session = null;
            ChannelSftp sftp = null;
            final List<Future<?>> futures = new ArrayList<>(list != null ? list.size() : 16);

            try {
                // === 建立 SFTP 连接（串行使用 ChannelSftp）===
                JSch jsch = new JSch();
                session = jsch.getSession(SFTP_USER, SFTP_HOST, SFTP_PORT);
                session.setPassword(SFTP_PASS);
                java.util.Properties cfg = new java.util.Properties();
                cfg.put("StrictHostKeyChecking", "no");
                session.setConfig(cfg);
                session.connect(10000);

                sftp = (ChannelSftp) session.openChannel("sftp");
                sftp.connect(10000);
                log.info("SFTP connected. baseDir={}", remoteBase);

                if (list == null || list.isEmpty()) {
                    log.info("No CDR candidates found. done. total=0, done=0, success=0, fail=0");
                    return "done. total=0, done=0, success=0, fail=0";
                }

                for (ECdrDOWithoutTag rec : list) {
                    String fileName = getRecordFileName(rec);

                    // Mongo 判重：过去 N 天已有语义识别则跳过
                    try {
                        String phone = CdrUtil.trimNumber(rec.getCalleetogatewaye164());
                        if (phone != null && !phone.isEmpty() && rp.skipDays > 0) {
                            if (hasParsedWithinDays(phone, rp.skipDays)) {
                                log.info("Skip by history ({}d parsed exists). phone={} file={}",
                                        rp.skipDays, phone, fileName);
                                maybeLogProgress(total, done.incrementAndGet(), ok.get(), fail.get());
                                continue;
                            }
                        }
                    } catch (Exception chkEx) {
                        log.warn("历史检查异常，继续处理。file={}", fileName, chkEx);
                    }

                    // 本次运行内去重：phone + dayKey
                    try {
                        String phone = CdrUtil.trimNumber(rec.getCalleetogatewaye164());
                        if (phone != null && !phone.isEmpty()) {
                            String dayKey = formatToPhYmd(rec.getStarttime());
                            String dedupKey = phone + "_" + dayKey;
                            if (!inRunSeen.add(dedupKey)) {
                                log.info("Skip duplicate within run: {}", dedupKey);
                                maybeLogProgress(total, done.incrementAndGet(), ok.get(), fail.get());
                                continue;
                            }
                        }
                    } catch (Exception ignore) { /* 兜底 */ }

                    try {
                        // ===== 串行 SFTP 定位&下载 =====
                        String remotePath = locateRemoteFileSftp(rec, sftp, remoteBase, fileName);
                        if (remotePath == null) {
                            log.warn("未找到远端文件：{}/**/{}", remoteBase, fileName);
                            fail.incrementAndGet();
                            maybeLogProgress(total, done.incrementAndGet(), ok.get(), fail.get());
                            continue;
                        }
                        Path localMp3 = mp3Dir.resolve(fileName);
                        downloadSftp(sftp, remotePath, localMp3.toFile());

                        // ===== 并发提交“转写→抽取→入库→清理”任务 =====
                        futures.add(pool.submit(() -> {
                            String base = FileUtil.mainName(fileName);
                            try {
                                // 转写
                                String transcript = aiNlpClient.transcribeAudioText(localMp3.toString());
                                log.info("转写结果 ({}):\n{}", base, transcript);

                                // 结构化抽取
                                String jsonStr = aiNlpClient.processTxtWithGpt(transcript);

                                boolean parsedOk = true;
                                try {
                                    JSONUtil.parseObj(jsonStr);
                                } catch (Exception jex) {
                                    parsedOk = false;
                                    log.warn("GPT 返回非严格对象 JSON，仍原样打印。file={}", fileName, jex);
                                }
                                log.info("GPT 结构化结果 ({}): {}", base, jsonStr);

                                // 写入 Mongo
                                upsertMongoRecord(rec, transcript, jsonStr, parsedOk, base);

                                ok.incrementAndGet();
                            } catch (Exception singleEx) {
                                log.error("处理失败：{}", fileName, singleEx);
                                fail.incrementAndGet();
                            } finally {
                                // 清理 mp3
                                try {
                                    Files.deleteIfExists(localMp3);
                                } catch (Exception delEx) {
                                    log.warn("删除本地 mp3 失败：{}", localMp3, delEx);
                                }
                                // 计入完成并按节距打印进度
                                maybeLogProgress(total, done.incrementAndGet(), ok.get(), fail.get());
                            }
                        }));
                    } catch (Exception e) {
                        log.error("SFTP 阶段失败：{}", fileName, e);
                        fail.incrementAndGet();
                        maybeLogProgress(total, done.incrementAndGet(), ok.get(), fail.get());
                    }
                }

            } catch (Exception e) {
                log.error("SFTP 执行异常", e);
                return "ERROR: " + e.getMessage();
            } finally {
                // 提交结束，关闭池 & 等待任务完成
                // 注意：这个 finally 是资源回收；互斥释放在外层 finally
                // 线程池
                // pool 可能还未初始化成功，这里判空
                // （但上面已赋值，仍做保护）
                if (pool != null) {
                    pool.shutdown();
                    try {
                        // 每分钟打一次进度，直到全部完成
                        while (!pool.awaitTermination(1, TimeUnit.MINUTES)) {
                            log.info("Processing... total={} done={} ok={} fail={}",
                                    total, done.get(), ok.get(), fail.get());
                        }
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        log.warn("等待线程池结束被中断");
                    }
                }

                // 确保 futures 中的异常都被显式取出（异常已在任务内计入 fail，这里仅避免静默）
                for (Future<?> f : futures) {
                    try { f.get(); } catch (Exception ignore) { /* 已在任务内处理 */ }
                }

                // 关闭 SFTP 连接
                // sftp / session 可能未建立成功，这里判空
                try {
                    if (sftp != null && sftp.isConnected()) sftp.disconnect();
                } catch (Exception ignore) {}
                try {
                    if (session != null && session.isConnected()) session.disconnect();
                } catch (Exception ignore) {}
            }

            String summary = String.format("done. total=%d, done=%d, success=%d, fail=%d",
                    total, done.get(), ok.get(), fail.get());
            log.info(summary);
            return summary;

        } finally {
            RUNNING.set(false);
            long durMs = System.currentTimeMillis() - t0;
            log.info("RecordingProcessingJob finished. lock released. elapsed={}ms", durMs);
        }
    }

    // ===================== 进度日志工具 =====================
    private static void maybeLogProgress(int total, int done, int ok, int fail) {
        if (total <= 0) return;
        if (done == total || done % PROGRESS_LOG_EVERY == 0) {
            log.info("Progress: total={} done={} ok={} fail={}", total, done, ok, fail);
        }
    }

    // ===================== SFTP 工具 =====================

    /**
     * SFTP：若 {remoteBase}/{HH}/{fileName} 存在则返回完整路径，否则 null
     */
    private static String locateRemoteFileSftp(ECdrDOWithoutTag rec,
                                               ChannelSftp sftp,
                                               String remoteBase,
                                               String fileName) {
        String tsStr = formatToPhYmdHms(rec.getStarttime()); // yyyyMMddHHmmss
        if (tsStr.length() != 14) {
            throw new IllegalArgumentException("formatToPhYmdHms 必须返回 yyyyMMddHHmmss，实际：" + tsStr);
        }
        String hourPart = tsStr.substring(8, 10); // HH
        String base = remoteBase.endsWith("/") ? remoteBase : remoteBase + "/";
        String fullPath = base + hourPart + "/" + fileName;
        try {
            SftpATTRS attrs = sftp.lstat(fullPath);
            if (attrs != null && !attrs.isDir()) {
                return fullPath;
            }
        } catch (SftpException ignore) {
            // 不存在
        }
        return null;
    }

    /**
     * SFTP：下载
     */
    private static void downloadSftp(ChannelSftp sftp, String remotePath, File localFile) throws SftpException {
        File parent = localFile.getParentFile();
        if (parent != null && !parent.exists()) parent.mkdirs();
        sftp.get(remotePath, localFile.getAbsolutePath());
    }

    // ===================== Mongo 写/查 =====================

    /**
     * 写入 Mongo：同一号码同一天数组 push 追加
     */
    private void upsertMongoRecord(ECdrDOWithoutTag rec, String transcript, String gptJson,
                                   boolean gptJsonOk, String baseFileName) {
        String phone = CdrUtil.trimNumber(rec.getCalleetogatewaye164());
        if (phone == null || phone.isEmpty()) {
            log.warn("被叫号码为空，跳过 Mongo 写入。rec={}", rec);
            return;
        }
        String dayKey = formatToPhYmd(rec.getStarttime());

        Document meta = toMetaDocument(rec, baseFileName);

        Document item = new Document("meta", meta)
                .append("transcript", transcript);
        if (gptJsonOk) {
            try {
                item.append("parsed", Document.parse(gptJson));
            } catch (Exception pe) {
                item.append("parsed_raw", gptJson);
            }
        } else {
            item.append("parsed_raw", gptJson);
        }

        MongoCollection<Document> coll = mongoTemplate.getDb().getCollection(MONGO_COLLECTION_NAME);

        Document query = new Document("_id", phone);
        Document update = new Document("$push", new Document("records." + dayKey, item))
                .append("$set", new Document("phone_md5", md5(phone))
                        .append("phone_md5+63", md5("+63" + phone)));

        coll.updateOne(query, update, new UpdateOptions().upsert(true));
        log.info("Mongo upsert OK. phone={} day={} file={}", phone, dayKey, baseFileName);
    }

    /**
     * 元信息转子文档
     */
    private static Document toMetaDocument(ECdrDOWithoutTag r, String fileName) {
        return new Document("callergatewayid", r.getCallergatewayid())
                .append("calleetogatewaye164", r.getCalleetogatewaye164())
                .append("calleeaccesse164", r.getCalleeaccesse164())
                .append("starttime", r.getStarttime())
                .append("stoptime", r.getStoptime())
                .append("holdtime", r.getHoldtime())
                .append("enddirection", r.getEnddirection())
                .append("callere164", r.getCallere164())
                .append("calleee164", r.getCalleee164())
                .append("callerip", r.getCallerip())
                .append("calleeip", r.getCalleeip())
                .append("flowno", r.getFlowno())
                .append("fileName", fileName);
    }

    /**
     * md5 工具
     */
    private static String md5(String input) {
        if (input == null) return "";
        try {
            MessageDigest md = MessageDigest.getInstance("MD5");
            byte[] messageDigest = md.digest(input.getBytes());
            StringBuilder sb = new StringBuilder();
            for (byte b : messageDigest) sb.append(String.format("%02x", b));
            return sb.toString();
        } catch (NoSuchAlgorithmException e) {
            return input;
        }
    }

    /**
     * 过去 N 天是否已有语义识别（parsed / parsed_raw）
     */
    private boolean hasParsedWithinDays(String phone, int days) {
        if (days <= 0) return false;
        MongoCollection<Document> coll = mongoTemplate.getDb().getCollection(MONGO_COLLECTION_NAME);

        List<String> dayKeys = dayKeysForLookback(days);
        Document projection = new Document("_id", 1);
        for (String k : dayKeys) projection.append("records." + k, 1);

        Document doc = coll.find(new Document("_id", phone)).projection(projection).first();
        if (doc == null) return false;

        Object recsObj = doc.get("records");
        if (!(recsObj instanceof Document)) return false;
        Document recs = (Document) recsObj;

        for (String k : dayKeys) {
            Object arr = recs.get(k);
            if (arr instanceof List) {
                @SuppressWarnings("unchecked")
                List<Object> l = (List<Object>) arr;
                for (Object o : l) {
                    if (o instanceof Document) {
                        Document it = (Document) o;
                        if (it.containsKey("parsed") || it.containsKey("parsed_raw")) {
                            return true;
                        }
                    }
                }
            }
        }
        return false;
    }

    /**
     * 生成回看天数的 yyyyMMdd 列表（含今日）
     */
    private static List<String> dayKeysForLookback(int days) {
        List<String> keys = new ArrayList<>(Math.max(days, 1));
        LocalDate today = LocalDate.now(CdrUtil.ZONE);
        for (int i = 0; i < days; i++) {
            keys.add(today.minusDays(i).format(DAY_FMT));
        }
        return keys;
    }

    // ============== 仅 KV（QueryString 风格）：skipDays/holdSeconds/limit/appointedDate ==============

    // RecordingProcessingJob 内部
    private static class RunParams {
        int skipDays = 7;      // 过去 N 天（含今日）已有语义识别则跳过
        int holdSeconds = 60;  // 通话时长阈值（秒）
        int limit = 0;         // 0 或未传 = 不限（不加 LIMIT）
        String appointedDate = "";  // 指定日期 yyyyMMdd（若有则用它的前一天）
    }

    /**
     * 仅识别 skipDays / holdSeconds / limit / appointedDate；未传 limit => 不限
     */
    private static RunParams parseParams(String param) {
        RunParams rp = new RunParams();
        if (param == null || param.trim().isEmpty()) return rp;

        String normalized = param.trim();
        if (normalized.startsWith("?")) normalized = normalized.substring(1);
        normalized = normalized.replace(';', '&').replace(',', '&');

        String[] parts = normalized.split("&");
        for (String part : parts) {
            String s = part.trim();
            if (s.isEmpty()) continue;
            int idx = s.indexOf('=');
            if (idx <= 0 || idx >= s.length() - 1) continue;

            String key = s.substring(0, idx).trim();
            String val = s.substring(idx + 1).trim();
            if (val.isEmpty()) continue;

            try {
                if ("skipDays".equalsIgnoreCase(key)) {
                    int n = Integer.parseInt(val);
                    if (n >= 0) rp.skipDays = n;
                } else if ("holdSeconds".equalsIgnoreCase(key)) {
                    int n = Integer.parseInt(val);
                    if (n >= 0) rp.holdSeconds = n;
                } else if ("limit".equalsIgnoreCase(key)) {
                    int n = Integer.parseInt(val);
                    // 仅当 >0 时生效；否则保持 0=不限
                    if (n > 0) rp.limit = n;
                } else if ("appointedDate".equalsIgnoreCase(key)) {
                    if (val.matches("\\d{8}")) {
                        rp.appointedDate = val;
                    } else {
                        log.warn("appointedDate 非法: {}", val);
                    }
                }
            } catch (NumberFormatException nfe) {
                log.warn("Invalid number for key {}: {}", key, val);
            }
        }
        return rp;
    }

    /**
     * 远端目录使用的 day：appointedDate(yyyyMMdd) 存在 -> appointedDate-1；否则昨天
     */
    private static String resolveTargetDayForBaseDir(String appointedDate) {
        LocalDate base;
        if (appointedDate != null && !appointedDate.isEmpty()) {
            try {
                base = LocalDate.parse(appointedDate, DAY_FMT);
            } catch (Exception e) {
                base = LocalDate.now(ZONE);
                log.warn("appointedDate 非法({})，按今天计算", appointedDate);
            }
        } else {
            base = LocalDate.now(ZONE);
        }
        return base.minusDays(1).format(DAY_FMT);
    }

    // 构造标准文件名
    private static String getRecordFileName(ECdrDOWithoutTag cdrRecordingDO) {
        StringBuilder sb = new StringBuilder();
        sb.append(formatToPhYmdHms(cdrRecordingDO.getStarttime())).append("-");
        sb.append(cdrRecordingDO.getCallere164()).append("_1").append("-");
        sb.append(cdrRecordingDO.getCalleee164()).append("_1").append("-");
        sb.append(cdrRecordingDO.getCallerip()).append("-");
        sb.append(cdrRecordingDO.getCalleeip()).append("-");
        sb.append(cdrRecordingDO.getFlowno()).append(".mp3");
        return sb.toString();
    }

    /**
     * 确保本地目录存在
     */
    private static void ensureLocalDir(String dir) {
        File f = new File(dir);
        if (!f.exists() && !f.mkdirs()) {
            throw new IllegalStateException("创建本地目录失败: " + dir);
        }
    }
}
